package com.example.storm.storm;

import com.alibaba.fastjson.JSON;
import com.example.storm.bean.InfoGet;
import com.example.storm.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Slf4j
public class Spout extends BaseRichSpout {

	private static final long serialVersionUID = -2548451744178936478L;
	private SpoutOutputCollector collector;

	@Override
	public void open(Map map, TopologyContext arg1, SpoutOutputCollector collector) {
		this.collector = collector;
	}
	
	@Override
	public void nextTuple() {
		while(true){
			try {
				//此处需要获取数据，可以通过mq poll获取数据，也可以通过db查询获取数据
				List<InfoGet> list = new ArrayList<>();
				InfoGet infoGet = new InfoGet();
				infoGet.setAge(25);
				infoGet.setName("void");
				list.add(infoGet);

				String sendMessage = JSON.toJSONString(list);
				log.info("Spout发送的消息：{}", sendMessage);
				//发送到bolt中
				this.collector.emit(new Values(sendMessage));

				//没有数据则休眠一段时间
				TimeUnit.SECONDS.sleep(10);
			} catch (Exception e) {
				log.error("Spout处理异常休眠10s!", e);
				try {
					TimeUnit.SECONDS.sleep(10);
				} catch (InterruptedException e1) {
					log.error("Spout暂停失败!",e1);
				}
			}
		}
	}

	/**
	 * 输出格式，对应bolt中获取
	 * @param declarer
	 */
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields(Constants.FIELD));
	}
}
